package com.chukun.flink.stream.state.key;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Map;

/**
 * @author chukun
 * @version 1.0.0
 * @description 带状态的flatmap操作符 MapStateFlatMap的基本使用
 * @createTime 2022年05月15日 22:59:00
 */
public class MapStateFlatMap extends RichFlatMapFunction<Tuple2<Integer, Integer>, Tuple2<Integer,Integer>> {

    private static final Logger logger = LoggerFactory.getLogger(MapStateFlatMap.class);

    // 创建MapState类型的状态结构
    private transient MapState<Integer, Integer> mapState;

    @Override
    public void flatMap(Tuple2<Integer, Integer> input, Collector<Tuple2<Integer, Integer>> collector) throws Exception {
        mapState.put(input.f0, input.f1);
        Iterator<Map.Entry<Integer, Integer>> iterator = mapState.iterator();
        logger.info("..........................");
        while (iterator.hasNext()) {
            Map.Entry<Integer, Integer> map = iterator.next();
            logger.info("{},{} ... {}", Thread.currentThread().getName(), map.getKey(), map.getValue());
        }
        collector.collect(input);
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        logger.info("{},{}", Thread.currentThread().getName(), "恢复或初始化状态");

        // 创建一个MapStateDescriptor状态描述符，指定名称为valueStateFlatMap
        MapStateDescriptor<Integer,Integer> descriptor = new MapStateDescriptor<>(
                "MapStateFlatMap", Integer.class, Integer.class);
        // 初始化mapState
        mapState = getRuntimeContext().getMapState(descriptor);
        logger.info("初始化 mapState ： {}", mapState);
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 封装了数据源函数，以及keyBy操作的数据流
        KeyedStream<Tuple2<Integer, Integer>, Integer> keyedStream = KeyStateBase.before(env);

        // 在keyedStream中使用有状态的FlatMap操作符
        DataStream<Tuple2<Integer, Integer>> resultStream = keyedStream.flatMap(new MapStateFlatMap());

        resultStream.print("输出结果 : ");

        env.execute("MapStateFlatMap");
    }
}
